package com.ruoyi.eqmonitor.dispose.service;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.ruoyi.common.redis.service.RedisService;
import com.ruoyi.eqmonitor.domain.OptDevice;
import com.ruoyi.eqmonitor.dtos.OptDeviceDto;
import com.ruoyi.eqmonitor.service.impl.OptDeviceServiceImpl;
import com.starwsn.protocol.core.Starwsn;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.ruoyi.eqmonitor.utils.Constant.MQTT_HARDWARE_TYPES;

/**
 * mqtt消息订阅服务
 *
 * @author dh
 */
@Slf4j(topic = "monitor_mqtt")
@Component
public class MqttService {
    private static final long TIME_INTERVAL = 15 * 60 * 1000;
    private Timer timer = null;
    private static Map<String, OptDeviceDto> subscribeMapping = new HashMap<>();

    @Value("${mqtt.host:tcp://112.102.225.89:18833}")
    private String host;

    @Value("${mqtt.clientid}:mqtt-yjbz-2024")
    private String clientid;

    @Value("${mqtt.username:admin}")
    private String username;

    @Value("${mqtt.password:123456}")
    private String password;

    @Value("${mqtt.keepAlive:10}")
    private int keepAlive;

    @Value("${mqtt.keepAliveInterval:20}")
    private int keepAliveInterval;

    @Value("${mqtt.enable:false}")
    private boolean enableMqtt;

    /** 定义每100个设备重新拉取一次订阅 */
    private static final int SPLIT_SIZE = 100;

    @Resource
    private AsyncConsumeMsg asyncConsumeMsg;

    @Resource
    private RedisService redisService;

    private static MqttClient client = null;

    /**
     * 前置处理 连接mqtt
     */
    @PostConstruct
    private void init() {
        if (enableMqtt) {
            connect(true);
        }
    }

    /**
     * 后置处理 销毁时关闭mqtt连接
     */
    @PreDestroy
    public void destroy() {
        disConnect();
    }

    /**
     * mqtt连接
     * @param cleanSession 是否清空之前连接的所有会话
     */
    public boolean connect(boolean cleanSession) {
        try {
            // host为主机名，test为clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
            client = new MqttClient(host, clientid, new MemoryPersistence());
            // MQTT的连接设置
            MqttConnectOptions options = new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
            options.setCleanSession(cleanSession);
            // 设置连接的用户名
            options.setUserName(username);
            // 设置连接的密码
            options.setPassword(password.toCharArray());
            // 设置超时时间 单位为秒
            options.setConnectionTimeout(keepAlive);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(keepAliveInterval);
            // 设置回调函数
            client.setCallback(new MqttCallback() {

                /** 连接失败的重试策略 */
                @Override
                public void connectionLost(Throwable throwable) {
                    log.warn("mqtt服务断开连接！！！");
                    for (int i = 0; i < 10000; i++) {
                        log.warn("第【" + (i + 1) + "】次尝试连接MQTT服务");
                        boolean cb = connect(false);
                        if (cb) {
                            log.warn("mqtt服务重连成功！！！");
                            /*
                             * 设备重连成功后需重新订阅所有设备消息主题
                             * */
                            subScription();
                            break;
                        }
                        try {
                            if (i < 5) {
                                Thread.sleep(5000);
                            } else if (i >= 5 && i < 10) {
                                Thread.sleep(10000);
                            } else {
                                Thread.sleep(60000);
                            }
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        if (i == 9999) {
                            log.warn("mqtt服务重连失败！！！");
                        }
                    }
                }

                /** 接收消息 */
                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    // 异步消费
                    asyncConsumeMsg.messageArrived(topic, message);
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                }
            });

            client.connect(options);

            log.info("mqtt服务连接成功！！！");
            return true;

        } catch (Exception e) {
            log.error("mqtt服务连接失败！！！", e);
            return false;
        }
    }

    /**
     * 断开客户端连接
     */
    public boolean disConnect() {
        try {
            client.disconnect();
            client.close();
            log.info("mqtt服务断开成功！！！");
            return true;
        } catch (MqttException e) {
            log.error("mqtt服务断开失败", e);
            return false;
        }
    }

    /**
     * 初始化调用订阅所有设备 禁止调用
     */
    @Deprecated
    public boolean subScription() {
        log.info("初始化调用订阅所有设备 禁止调用");
        Map<String, OptDeviceDto> deviceList = redisService.getCacheMap(OptDeviceServiceImpl.REAL_DEVICE_REDIS_KEY);
        //Map<String, OptDeviceDto> deviceList = new HashMap<>();
        if (CollectionUtil.isEmpty(deviceList)) return false;
        String[] topics = deviceList.values().stream()
                .map(device -> formatTopic(device.getRealDeviceCode(), device.getAccessType()))
                .filter(ObjectUtil::isNotNull)
                .toArray(String[]::new);
        // 记录一下订阅的设备
        subscribeMapping = deviceList.values().stream().collect(Collectors.toMap(OptDeviceDto::getDeviceCode, Function.identity()));
        // 每次最多订阅100
        if (topics.length <= SPLIT_SIZE) {
            log.info("每次最多订阅100");
            return subScription(topics);
        } else {
            // 批量订阅
//            for(int j=0; j < topics.length - ( topics.length % SPLIT_SIZE ); j += SPLIT_SIZE ){
//                String[] subtopics= Arrays.copyOfRange(topics, j, j + SPLIT_SIZE);
//                subScription(subtopics);
//            }
//            // 剩余订阅
//            String[] subtopics = Arrays.copyOfRange(topics,topics.length-topics.length % SPLIT_SIZE,topics.length);
//            return subScription(subtopics);
            boolean flag = true;
            List<List<String>> split = ListUtil.split(ListUtil.toList(topics), SPLIT_SIZE);
            for (List<String> strings : split) {
                flag = flag && subScription(strings.toArray(new String[0]));
            }
            return flag;
        }
    }


    /**
     * 主题订阅
     * @param snCode 设备编码
     */
    public boolean subScription(String snCode, String hardwareType) {
        String topic = formatTopic(snCode, hardwareType);
        if (topic == null) {
            return true;
        }
        return this.subScription(new String[]{topic});
    }

    /**
     * 主题订阅
     * @param topics 主题名称数组
     */
    public boolean subScription(String[] topics) {
        String topicStr = ArrayUtil.join(topics, ",");
        if (ArrayUtil.isEmpty(topics)) {
            log.error("mqtt主题【{}】订阅失败！！！原因：主题内容为空", topicStr);
            return false;
        }

        //订阅泵站
        String[] finalTopics = new String[topics.length+1];
        System.arraycopy(topics, 0, finalTopics, 0, topics.length);
        finalTopics[topics.length] = "data_list_bigData/bz_data_json";
        try {
            // 订阅消息
           // String[] tempTopics = new String[1];
            client.subscribe(finalTopics);
            log.info("mqtt主题【{}】:【{}】订阅成功！！！", finalTopics.length, ArrayUtil.join(finalTopics, ","));
            return true;
        } catch (Exception e) {
            log.error("mqtt主题【{}】订阅失败！！！原因：{}", topicStr, e);
            return false;
        }
    }

    /**
     * 取消主题订阅
     */
    public boolean unSubScription(String snCode, String hardwareType) {
        String topic = formatTopic(snCode, hardwareType);
        if (topic == null) {
            return true;
        }
        return this.unSubScription(new String[]{topic});
    }

    /**
     * 取消主题订阅
     *
     * @param equipmentInfos 设备对象
     */
    public boolean unSubScription(List<Map<String, Object>> equipmentInfos) {
        String[] topics = equipmentInfos.stream()
                .map(info -> formatTopic(MapUtil.getStr(info, "snCode"), MapUtil.getStr(info, "hardwareType")))
                .filter(ObjectUtil::isNotNull)
                .toArray(String[]::new);
        return this.unSubScription(topics);
    }

    /**
     * 取消主题订阅
     *
     * @param topics 主题名称
     */
    public boolean unSubScription(String[] topics) {
        String topic = "";
        try {
            for (String s : topics) {
                topic = s;
                client.unsubscribe(topic);
                log.warn("mqtt主题{" + topic + "}取消订阅成功！！！");
            }
            return true;
        } catch (Exception e) {
            log.warn("mqtt主题{" + topic + "}取消订阅失败！！！");
            return false;
        }
    }

    /**
     * 格式化mqtt主题
     * @param snCode 设备编码
     * @param hardwareType 硬件类型
     * @return 格式化后的mqtt主题
     */
    private String formatTopic(String snCode, String hardwareType) {
        if (!ArrayUtil.contains(MQTT_HARDWARE_TYPES, hardwareType)) {
            // 校验一级主题(GW 或 ND)
            log.warn("一级主题名" + hardwareType + "不合法");
            return null;
        }
        if (StrUtil.isEmpty(snCode) || snCode.length() < 6) {
            // 校验二级主题 (6位snCode编码)
            log.warn("二级主题名" + snCode + "不合法");
            return null;
        }
        return String.format("%s/%s/Report", "Data", snCode);
    }

    public String bytesToHexString(byte[] src) {
        StringBuilder stringBuilder = new StringBuilder("");
        if (src != null && src.length > 0) {
            byte[] arr$ = src;
            int len$ = src.length;

            for (int i$ = 0; i$ < len$; ++i$) {
                byte b = arr$[i$];
                int v = b & 255;
                String hv = Integer.toHexString(v);
                if (hv.length() < 2) {
                    stringBuilder.append(0);
                }

                stringBuilder.append(hv);
            }

            return stringBuilder.toString();
        } else {
            return null;
        }
    }

    /**
     * 推送消息
     * @param topicName  主题名称
     * @param payload  消息内容
     */
    public boolean publish(String topicName, byte[] payload){
        try {
            MqttTopic topic = client.getTopic(topicName);
            MqttMessage message = new MqttMessage();
            message.setQos(1);
            message.setRetained(false);
            message.setPayload(payload);
            MqttDeliveryToken token = topic.publish(message);
            log.info("mqtt消息发送结果等待。。。。。");
            token.waitForCompletion();
            log.info("mqtt消息发送结果："+ token.isComplete());
            return token.isComplete();
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 启动监听设备编码变更
     */
    public void listenDeviceCodeChange() {
        if (!enableMqtt) {
            return;
        }
        if (timer != null) {
            timer.cancel();
        }
        // 定义一个每15分钟执行一次的定时任务
        timer = new Timer(true);
        // 15分钟后开始执行 之后每隔15分钟执行一次
        timer.schedule(new DeviceCodeChangeListener(), TIME_INTERVAL, TIME_INTERVAL);
    }

    /**
     * 设备编码变更监听器
     */
    public class DeviceCodeChangeListener extends TimerTask {
        @Override
        public void run() {
            log.info("开始检查MQTT设备编码变更");
            try {
                Map<String, OptDeviceDto> deviceList = redisService.getCacheMap(OptDeviceServiceImpl.REAL_DEVICE_REDIS_KEY);
                boolean isCurrentEmpty = MapUtil.isEmpty(deviceList);
                boolean isSubscribeEmpty = MapUtil.isEmpty(subscribeMapping);
                if (isCurrentEmpty && isSubscribeEmpty) {
                    return;
                }

                // 两个设备列表长度不一致 则一定被改变了
                boolean needReSubscribe = deviceList.size() != subscribeMapping.size();
                if (needReSubscribe) {
                    log.info("设备数量发生变更，重新订阅MQTT, 原设备数量：{}，新设备数量：{}", subscribeMapping.size(), deviceList.size());
                }
                // 检查一下明细有没有变更
                if (!needReSubscribe) {
                    for (Map.Entry<String, OptDeviceDto> entry : subscribeMapping.entrySet()) {
                        OptDeviceDto device = deviceList.get(entry.getKey());
                        if (device == null) {
                            log.info("设备编码未找到，重新订阅MQTT, 设备编码: {}", entry.getKey());
                            needReSubscribe = true;
                            break;
                        }
                        OptDeviceDto value = entry.getValue();
                        boolean deviceCodeEq = StrUtil.equals(value.getRealDeviceCode(), device.getRealDeviceCode());
                        boolean accessTypeEq = StrUtil.equals(value.getAccessType(), device.getAccessType());
                        if (!deviceCodeEq || !accessTypeEq) {
                            log.info("设备编码发生变更，重新订阅MQTT, 原设备编码：{}，新设备编码：{}，原accessType：{}，新原accessType：{}",
                                    value.getRealDeviceCode(), device.getRealDeviceCode(), value.getAccessType(), device.getAccessType());
                            needReSubscribe = true;
                            break;
                        }
                    }
                }
                if (needReSubscribe) {
                    log.info("设备编码发生变更，重新订阅MQTT");


                    boolean disConnectFlag = disConnect();
                    if (!disConnectFlag) {
                        log.warn("MQTT断开连接失败 准备尝试重新断开");
                        for (int i = 1; i <= 10; i++) {
                            log.warn("第" + i + "次尝试断开MQTT连接");
                            disConnectFlag = disConnect();
                            if (disConnectFlag) {
                                break;
                            }
                        }
                    }
                    if (!disConnectFlag) {
                        log.warn("尝试重连后 - MQTT断开连接失败");
                        return;
                    }


                    boolean connect = connect(true);
                    if (!connect) {
                        log.warn("MQTT连接失败 准备尝试重连");
                        for (int i = 0; i < 10; i++) {
                            log.warn("第" + i + "次尝试重连MQTT");
                            connect = connect(true);
                            if (connect) {
                                break;
                            }
                        }
                    }
                    if (!connect) {
                        log.warn("尝试重连后 - MQTT连接失败");
                        return;
                    }

                    String[] topics = deviceList.values().stream()
                            .map(device -> formatTopic(device.getRealDeviceCode(), device.getAccessType()))
                            .filter(ObjectUtil::isNotNull)
                            .toArray(String[]::new);
                    if (topics.length <= 0) {
                        log.warn("重新订阅MQTT失败, topic为空! {}", deviceList);
                    }
//                    // 记录一下订阅的设备
//                    subscribeMapping = deviceList.values().stream().collect(Collectors.toMap(OptDeviceDto::getDeviceCode, Function.identity()));
                    // 每次最多订阅100
                    if (topics.length <= SPLIT_SIZE) {
                        boolean subScriptionFlag = subScriptionAndRetry(topics);
                        if (!subScriptionFlag) {
                            log.warn("尝试重新订阅后 - MQTT订阅失败, topics: {}", ListUtil.toList(topics));
                            return;
                        }
                    } else {
//                        // 批量订阅
//                        for(int j=0; j < topics.length - ( topics.length % SPLIT_SIZE ); j += SPLIT_SIZE ){
//                            String[] subtopics= Arrays.copyOfRange(topics, j, j + SPLIT_SIZE);
//                            boolean subScriptionFlag = subScriptionAndRetry(subtopics);
//                            if (!subScriptionFlag) {
//                                log.warn("尝试重新订阅后 - MQTT订阅失败, topics: {}", ListUtil.toList(topics));
//                                return;
//                            }
//                        }
//                        // 剩余订阅
//                        String[] subtopics = Arrays.copyOfRange(topics,topics.length-topics.length % SPLIT_SIZE,topics.length);
//                        boolean subScriptionFlag = subScriptionAndRetry(subtopics);
//                        if (!subScriptionFlag) {
//                            log.warn("尝试重新订阅后 - MQTT订阅失败, topics: {}", ListUtil.toList(topics));
//                            return;
//                        }
                        boolean flag = true;
                        List<List<String>> split = ListUtil.split(ListUtil.toList(topics), SPLIT_SIZE);
                        for (List<String> strings : split) {
                            flag = flag && subScription(strings.toArray(new String[0]));
                        }
                        if (!flag) {
                            log.warn("尝试重新订阅后 - MQTT订阅失败, topics: {}", ListUtil.toList(topics));
                            return;
                        }
                    }
                    log.info("重新订阅MQTT完成, 订阅数量: {}", topics.length);
                }
                subscribeMapping = deviceList;
            } catch (Exception e) {
                log.warn("重新订阅MQTT失败", e);
            }
            log.info("开始检查MQTT设备编码变更  -- 完成!");
        }
    }

    private boolean subScriptionAndRetry( String[] topics) {
        boolean subScriptionFlag = subScription(topics);
        if (!subScriptionFlag) {
            log.warn("MQTT订阅失败 准备尝试重新订阅");
            for (int i = 1; i <= 10; i++) {
                log.warn("第" + i + "次尝试订阅MQTT");
                subScriptionFlag = subScription(topics);
                if (subScriptionFlag) {
                    break;
                }
            }
        }
        if (!subScriptionFlag) {
            log.warn("尝试重新订阅后 - MQTT订阅失败, topics: {}", ListUtil.toList(topics));
            return false;
        }
        return true;
    }
}
